#pragma once

#include "../common/Log.hpp"
#include "../common/SqliteHelper.hpp"
#include "../common/FileHelper.hpp"
#include "../common/UUIDHelper.hpp"
#include "../common/message.pb.h"
#include "MessageMapper.hpp"
#include <iostream>
#include <string>
#include <list>
#include <unordered_map>
#include <mutex>
#include <memory>


namespace micromq
{
    class MessageCenter
    {
    public:
        using ptr = std::shared_ptr<MessageCenter>;
    public:
        MessageCenter(std::string& baseDir, const std::string& msgQueueName)
            : _mapper(baseDir, msgQueueName), _msgQueueName(msgQueueName), _validDurableNum(0), _totalDurableNum(0)
        {}

        void recovery()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            _waitConsumeMsgs = _mapper.loadValidMessages();
            for(auto& msg : _waitConsumeMsgs)
            {
                _durableMsgs[msg->payload().properties().id()] = msg;
            }
            _validDurableNum = _waitConsumeMsgs.size();
            _totalDurableNum = _waitConsumeMsgs.size();
        }

        void addMessage(const BasicProperties* bp, const std::string& content, bool msgQueueDurable)
        {
            MessagePtr msg = std::make_shared<Message>();
            if(bp != nullptr)
            {
                DeliveryMode mode = msgQueueDurable ? bp->mode() : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(bp->id());
                msg->mutable_payload()->mutable_properties()->set_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            }
            else
            {
                DeliveryMode mode = msgQueueDurable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE;
                msg->mutable_payload()->mutable_properties()->set_id(UUIDHelper::getUUID());
                msg->mutable_payload()->mutable_properties()->set_mode(mode);
                msg->mutable_payload()->mutable_properties()->set_routing_key("");
            }
            msg->mutable_payload()->set_content(content);

            {
                std::unique_lock<std::mutex> lock(_mtx);
                if(msg->payload().properties().mode() == DeliveryMode::DURABLE)
                {
                    msg->mutable_payload()->set_valid("1");
                    _mapper.writeMessage(msg);
                    _durableMsgs[msg->payload().properties().id()] = msg;
                    _validDurableNum++;
                    _totalDurableNum++;
                }
                _waitConsumeMsgs.push_back(msg); 
            }
        }

        void removeMessage(const std::string& msgId)
        {
            std::unique_lock<std::mutex> lock(_mtx);
            auto it = _waitAckMsgs.find(msgId);
            if(it == _waitAckMsgs.end()) return;
            if(it->second->payload().properties().mode() == DeliveryMode::DURABLE)
            {
                _mapper.eraseMessage(it->second);
                _durableMsgs.erase(msgId);
                _validDurableNum--;
                garbageCollection();
            }
            _waitAckMsgs.erase(msgId);
        }

        MessagePtr getMessage()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            if(_waitConsumeMsgs.size() == 0) return nullptr;
            MessagePtr msg = _waitConsumeMsgs.front();
            _waitConsumeMsgs.pop_front();
            _waitAckMsgs[msg->payload().properties().id()] = msg;
            return msg;
        }

        std::list<MessagePtr> getWaitConsumeMessages()
        {
            return _waitConsumeMsgs;
        }

        int getWaitConsumeNum()
        {
            return _waitConsumeMsgs.size();
        }

        int getWaitAckNum()
        {
            return _waitAckMsgs.size();
        }

        int getValidDurableNum()
        {
            return _durableMsgs.size();
        }

        int getTotalDurableNum()
        {
            return _totalDurableNum;
        }

        void clear()
        {
            std::unique_lock<std::mutex> lock(_mtx);
            _mapper.removeMsgFile();
            _waitConsumeMsgs.clear();
            _waitAckMsgs.clear();
            _durableMsgs.clear();
            _validDurableNum = 0;
            _totalDurableNum = 0;
        }

        ~MessageCenter()
        {}
    private:
        void garbageCollection()
        {
            // 持久化的总消息数大于2000，且有效消息比例小于50%，则进行垃圾回收
            if(_totalDurableNum < 2000 || _validDurableNum * 10 / _totalDurableNum >= 5) return;
            std::list<MessagePtr> msgList = _mapper.loadValidMessages();
            for(auto& a : msgList)
            {
                MessagePtr msg = _durableMsgs[a->payload().properties().id()];
                msg->set_offset(a->offset());
                msg->set_length(a->length()); //更新为垃圾回收后在文件中的位置
            }
            _validDurableNum = msgList.size();
            _totalDurableNum = msgList.size();
        }
    private:
        std::mutex _mtx;
        MessageMapper _mapper;
        std::string _msgQueueName;
        int _validDurableNum;
        int _totalDurableNum;
        std::list<MessagePtr> _waitConsumeMsgs;
        std::unordered_map<std::string, MessagePtr> _waitAckMsgs; // <消息id, 消息指针>
        std::unordered_map<std::string, MessagePtr> _durableMsgs;
    };
}














/**
 * 构造函数是否有必要加锁问题：
 * C++对象在构造函数调用，开始分配内存就已经逐步生成，对其他线程可见，
 * 其他线程便可与构造函数并发访问成员变量，此时成员变量被多线程重入读写，线程不安全，
 * 而一个对象如果在构造函数初始化列表执行期间就可能被多线程并发访问，即使加锁，也要等到执行到构造函数主体时才能加，
 * 此时想保证线程安全，只能由这个对象的构造者保证，本类加锁无意义
 * 
 * 保证线程安全方案：
 * 1.使用智能指针，智能指针在创建对象时是线程局部化的，对象构造不完整，其他线程无法访问
 * 2.使用工厂模式，在工厂函数中加锁，即可保证同步
 * 3.延迟初始化，将初始化方法单独写成函数，函数内加锁，对象构造完毕后，再手动调用
 * 
 * 因此：
 * MessageCenter的线程安全，应由MessageManager的智能指针保证，构造函数不加锁
*/